package com.shengzai.flink.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

object Demo6MySQLSource {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val mysqlDS: DataStream[String] = env.addSource(new mysqlSource)
    mysqlDS.print()

    env.execute()

  }

}
class mysqlSource extends SourceFunction[String]{
  var conn: Connection = _
  var stat: PreparedStatement =_
  var res: ResultSet =_

  override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
    Class.forName("com.mysql.jdbc.Driver")
     conn = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata22?useUnicode=true&characterEncoding=UTF-8", "root", "123456")
     stat = conn.prepareStatement("select * from student2")
     res = stat.executeQuery()
    while (res.next()){
      val sid: String = res.getString(1)
      val name: String = res.getString(2)
      val age: Int = res.getInt(3)
      val sex: String = res.getString(4)
      val clazz: String = res.getString(5)
      ctx.collect(s"$sid\t$name\t$age\t$sex\t$clazz")
      Thread.sleep(500)
    }
  }

  override def cancel(): Unit = {
  conn.close()
  }
}
